#pragma once

#ifndef __linux__
#error "epoll_reactor.hpp only support linux"
#endif // __linux__

#include <sys/epoll.h>

CRANE_NS_BEGIN
namespace aio {


struct Reactor::Timer {
    int64_t             expiry;
    IReactorCallback*   callback;
    bool                active {true};  // 是否有效, 如果是false，则wait方法取到后，将其删除

    bool operator<(const Timer& rhs) const {
        return expiry > rhs.expiry;    // 现在需要小的优先
    }
};


struct ReactorImpl {
    fd_t                        epfd { CRANE_INVALID_FD };
    size_t                      event_count { 0 };  // 当前注册的事件数
    std::vector<epoll_event>    revents;
    
    std::priority_queue<Timer>  timers;         // 计时器
    
    static int ops_to_events(int events);
    static int events_to_ops(int events);

}; // ReactorImpl




int ReactorImpl::ops_to_events(int ops)
{
    int events = ReactorOps::OpNone;

    if (ops & ReactorOps::OpRead) events |= EPOLLIN;
    if (ops & ReactorOps::OpWrite) events |= EPOLLOUT;

    return events;
}

int ReactorImpl::events_to_ops(int events)
{
    int ops = ReactorOps::OpNone;

    if ( events & ReactorOps::OpRead )  ops |= OPS_READ;
    if ( events & ReactorOps::OpWrite ) ops |= OPS_WRITE;
    if ( events & ReactorOps::OpError ) ops |= OPS_ERROR;
    
    return ops;
}

CRANE_INLINE
void Reactor::init(Error & err)
{
    m_impl = std::make_shared<ReactorImpl>();
    m_impl->epfd = epoll_create1(EPOLL_CLOEXEC);
    if (m_impl->epfd == -1) err = CRANE_MAKE_SYSTEM_ERROR(errno, "epoll_create1");
}

CRANE_INLINE
void Reactor::add(fd_t fd, int ops, IReactorCallback *cb, Error & err)
{
    epoll_event ev;
    ev.events = ReactorImpl::ops_to_events(ops) | ;
    ev.data.ptr = cb;

    int r = ::epoll_ctl(m_impl->epfd, EPOLL_CTL_ADD, fd, &ev);
    if (r == -1) {
        err = CRANE_MAKE_SYSTEM_ERROR(errno, "epoll_ctl(ADD)");
        return;
    }
    m_impl->event_count += 1;
    return;
}

CRANE_INLINE
void Reactor::modify(fd_t fd, int ops, IReactorCallback *cb, Error & err)
{ 
    epoll_event ev;
    ev.events = ReactorImpl::ops_to_events(ops);
    ev.data.ptr = cb;

    int r = ::epoll_ctl(m_impl->epfd, EPOLL_CTL_MOD, fd, &ev);
    if (r == -1) {
        err = CRANE_MAKE_SYSTEM_ERROR(errno, "epoll_ctl(MOD)");
        return;
    }
    return;
}

CRANE_INLINE
void Reactor::remove(fd_t fd, Error & err)
{ 
    int r = ::epoll_ctl(m_impl->epfd, EPOLL_CTL_DEL, fd, NULL);
    if (r == -1) {
        err = CRANE_MAKE_SYSTEM_ERROR(errno, "epoll_ctl(DEL)");
        return;
    }
    m_impl->event_count -= 1;
    return;
}

CRANE_INLINE

void Reactor::wait(Error & err) { return wait(-1, err); }

CRANE_INLINE
void Reactor::wait(int timeout, Error & err)
{
    // 检查定时器队列，获取实际的超时时间.
    int tmout = timeout;
    int64_t now = CRANE_SYSTEM_TIMESTAMP_USEC();
    int64_t exp = timeout>=0?(now+(int64_t)timeout*1000):INT64_MAX;
    if ( !m_impl->m_timers.empty() ) {
        auto & timer = m_impl->m_timers.top();
        if ( timer.expiry < exp ) exp = timer.expiry;
    }

    if ( exp >= INT32_MAX )  tmout = -1;        // exp为无限等待
    else if ( exp < now ) tmout = 0;            // 已经超时
    else tmout = (int)(exp-now) / 1000LL;       // 剩余超时时间

    tmout = exp>INT32_MAX?-1:(int)((exp-now) / 1000LL);

    m_impl->revents.resize(m_impl->event_count);

    int r = ::epoll_wait(m_impl->epfd,
                         m_impl->revents.data(),
                         m_impl->revents.size(),
                         tmout);
    if (r == -1) {
        err = CRANE_MAKE_SYSTEM_ERROR(errno, "epoll_wait");
        return;
    }

    // 先处理实际事件
    for (int i = 0; i < r; ++i) {
        auto & ev = m_impl->revents[i];
        IReactorCallback * cb = (IReactorCallback*)ev.data.ptr;

        int ops = ReactorImpl::events_to_ops(ev.events);
        cb->onEvent(ops);
    }

    // 再处理超时，因为事件处理前可能有一批超时，事件处理后可能还有新增一批超时，现在可以一并处理
    now = CRANE_SYSTEM_TIMESTAMP_USEC();
    while ( !m_impl->m_timers.empty() ) {
        auto & timer = m_impl->m_timers.top();
        if ( timer.expiry < now ) break;
        
        // 先移除再回调，避免回调过程中再次添加定时器造成死锁或者列表错乱, 然后继续判断下一个节点是否超时
        auto cb = timer.callback;
        m_impl->m_timers.pop();
        cb->onEvent(ReactorOps::OpTimeout);
    }
    
    if ( r == 0 ) {
        err = CRANE_MAKE_ERROR(ErrorCode::eTimeout, "epoll_wait timeout");
    }

    return;
}


CRANE_INLINE 
Reactor::Timer * Reactor::createTimer(int64_t exp, IReactorCallback * cb, Error & err)
{ 
    std::unique_ptr<Timer> t(new Timer{exp, cb});
    Reactor::Timer * p = t.get();

    m_impl->timers.push(std::move(t));
    return p;
}



} // namespace aio
CRANE_NS_BEGIN
